package com.yindian.cos.logging.elastic;

import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.logging.log4j.nosql.appender.NoSqlConnection;
import org.apache.logging.log4j.nosql.appender.NoSqlObject;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.reflections.Reflections;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;

/**
 * log4j2获取nosql连接并输出到nosql
 *
 * @author zhangchuanhao
 * @date 2018/12/10
 */
@Log4j2
public class ElasticsearchConnection implements NoSqlConnection<Map<String, Object>, ElasticsearchObject> {

    private final String index;
    private final String type;
    private final RestHighLevelClient client;
    private final BulkProcessor bulkProcessor;
    private final AtomicBoolean closed = new AtomicBoolean(false);
    private LogOutputObject logOutputObject;

    public ElasticsearchConnection(String host,
                                   int port,
                                   String index,
                                   String type,
                                   int maxActionsPerBulkRequest,
                                   int maxConcurrentBulkRequests,
                                   ByteSizeValue maxVolumePerBulkRequest,
                                   TimeValue flushInterval,
                                   String packageName) {
        this.index = index;
        this.type = type;
        this.logOutputObject = init(packageName);

        // 初始化RestHighLevelClient
        RestClientBuilder restClientBuilder = RestClient.builder(new HttpHost(host, Integer.valueOf(port)));
        this.client = new RestHighLevelClient(restClientBuilder);

        // 初始化批量处理器
        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
//                int numberOfActions = request.numberOfActions();
//                log.debug("Executing bulk [{}] with {} requests",
//                        executionId, numberOfActions);
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                if (response.hasFailures()) {
                    log.warn("Bulk [{}] executed with failures", executionId);
                }
//                else {
//                    log.debug("Bulk [{}] completed in {} milliseconds",
//                            executionId, response.getTook().getMillis());
//                }
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                log.error("Failed to execute bulk", failure);
            }
        };

        BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer =
                (request, bulkListener) -> client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener);

        BulkProcessor.Builder builder = BulkProcessor.builder(bulkConsumer, listener);
        builder.setBulkActions(maxActionsPerBulkRequest);
        if (maxVolumePerBulkRequest != null) {
            builder.setBulkSize(maxVolumePerBulkRequest);
        }
        builder.setConcurrentRequests(maxConcurrentBulkRequests);

        builder.setFlushInterval(flushInterval);

        this.bulkProcessor = builder.build();
    }

    /**
     * 初始化LogOutputObject
     *
     * @param packageName LogOutputObject实现类所在的包，没有则传null
     * @return
     */
    private LogOutputObject init(String packageName) {
        try {
            return getAllSubclassOfTestInterface(packageName);
        } catch (Exception e) {
            throw new RuntimeException("Init LogOutputObject failed", e);
        }
    }

    /**
     * 获取LogOutputObject的所有实现类
     *
     * @return
     */
    private LogOutputObject getAllSubclassOfTestInterface(String logOutputObjectPackageName) throws Exception {
        if (StringUtils.isNotBlank(logOutputObjectPackageName)) {
            Reflections reflections = new Reflections(logOutputObjectPackageName);
            Set<Class<? extends LogOutputObject>> classes = reflections.getSubTypesOf(LogOutputObject.class);
            if (classes.size() > 0 && classes != null) {
                // 当前包名
                String packageName = ElasticsearchConnection.class.getPackage().getName();
                int i = 0;
                for (Class<? extends LogOutputObject> cls : classes) {
                    String nowClsPackageName = cls.getPackage().getName();
                    if (!packageName.equals(nowClsPackageName)) {
                        i++;
                    }
                }
                // 有多个实现类，抛出异常
                if (i > 1) {
                    throw new RuntimeException("There can only be one LogOutputObject implementation class but " + i + " are found");
                }
                // 只有一个实现类，直接初始化并返回
                else if (i == 1) {
                    for (Class<? extends LogOutputObject> cls : classes) {
                        String nowClsPackageName = cls.getPackage().getName();
                        if (!packageName.equals(nowClsPackageName)) {
                            return cls.newInstance();
                        }
                    }
                }
            }
        }
        return null;
    }

    @Override
    public ElasticsearchObject createObject() {
        return new ElasticsearchObject();
    }

    @Override
    public ElasticsearchObject[] createList(final int length) {
        return new ElasticsearchObject[length];
    }

    @Override
    public void insertObject(final NoSqlObject<Map<String, Object>> object) {
        if (logOutputObject == null) {
            logOutputObject = new LogOutputObject() {
            };
        }
        Map<String, Object> source = logOutputObject.outPutObject(object.unwrap());
        boolean isOutput = logOutputObject.isOutput(source);
        if (isOutput) {
            ElasticsearchUtil.index(bulkProcessor, source, index, type);
        }
    }

    @Override
    public void close() {
        closed.compareAndSet(false, true);
//        if (closed.compareAndSet(false, true)) {
//            ElasticsearchUtil.close(client, bulkProcessor);
//            ElasticsearchConnectionPool.closeElasticsearchConnection();
//        }
    }

    @Override
    public boolean isClosed() {
        return closed.get();
    }
}
